package com.concurrent.sample.business.event.manager.loan.impl;

import com.concurrent.sample.business.common.consts.BusinessConsts;
import com.concurrent.sample.business.event.manager.loan.ILoanDisruptorThreadService;
import com.concurrent.sample.business.event.manager.loan.ILoanEventService;
import com.concurrent.sample.business.event.param.OptionEvent;
import com.concurrent.sample.business.event.param.loan.LoanOptionEvent;
import java.util.HashMap;
import java.util.Objects;
import java.util.concurrent.ConcurrentLinkedQueue;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

@Slf4j
@Service
@RequiredArgsConstructor
public class LoanDisruptorThreadServiceImpl implements ILoanDisruptorThreadService, ApplicationRunner {

  private ConcurrentLinkedQueue<OptionEvent> queue = new ConcurrentLinkedQueue();

  private HashMap<Integer, LoanWorkThread> map = new HashMap<>();

  private final ILoanEventService loanEventService;

  /**
   * 监听约券事件
   * @param loanOptionEvent
   */
  @Override
  @Async(value = BusinessConsts.LOAN_EVENT_THREAD)
  @EventListener(LoanOptionEvent.class)
  public void listenerLoanOptionEvent(LoanOptionEvent loanOptionEvent) {
    push(loanOptionEvent.getOptionEvent());
  }

  @Override
  public void push(OptionEvent optionEvent) {
    if(Objects.nonNull(optionEvent)){
      queue.add(optionEvent);
    }
  }

  @Override
  public void run(ApplicationArguments args) throws Exception {
    // 初始化工作线程
    for (int i = 0; i < 10; i++) {
      LoanWorkThread loanWorkThread = new LoanWorkThread(loanEventService);
      loanWorkThread.start();
      map.put(i, loanWorkThread);
    }

    // TODO 待确认：无 Task 时等待多久，分发、工作线程各自的配置时间
    while(true){
      // TODO 异常捕获处理
      try{
        OptionEvent optionEvent = queue.poll();
        if(Objects.isNull(optionEvent)){
//        log.info("disruptor task is empty");
          Thread.sleep(200);
          continue;
        }
        distributeTask(optionEvent);
      } catch (Exception e) {
        log.error("disruptor task exception", e);
      }

    }
  }

  /**
   * 分发任务
   * @param optionEvent
   */
  private void distributeTask(OptionEvent optionEvent){
    String routeKey = optionEvent.getParam().getRouteKey();
    int workThreadKey = Math.abs(routeKey.hashCode() % map.size());
    map.get(workThreadKey).push(optionEvent);
  }

}
